#include <vector>
#include <map>
#include <logger.hpp>
#include "CmdHandler.h"

#include <algorithm>

#include <xbase_utility.hpp>
#include <compressor.hpp>

#include <table_mutator.hpp>
#include <row_mutation_data.hpp>

bool CInsertData_Compare(CInsertData* left_hand_side, CInsertData* right_hand_side)
{
	return left_hand_side->cid_ < right_hand_side->cid_;
}

void * CQueryCmdHandler::InsertThread(void * arg)
{
    LOG(DEBUG, "CQueryCmdHandler::InsertThread");
    //ASSERT(arg != 0,"invalid arg");
	
    CQueryCmdHandler &self = *(CQueryCmdHandler *)arg;

	// 用于多行写
	xbase::multi_row_mutator_ptr multi_row_mut = clientEnvironmentPtr_->new_multi_row_mutator();
	container_ptr container_p(new container_type);

	std::vector< CInsertData* > vec;
	vec.reserve(self.batWriteCount_ + 100);
	container_p->reserve(self.batWriteCount_ + 100);
	
    for (CInsertData *pData = 0;;)
	{
        bool ret = self.insertQue_.Pop(pData,self.insertWaitInterval_);
        if(ret)
		{
			vec.push_back(pData);
            if(vec.size() + container_p->size() >= self.batWriteCount_)
                self.WriteXbase(vec, multi_row_mut, container_p);
        }
		else
		{
            if(!vec.empty() || !container_p->empty())
                self.WriteXbase(vec, multi_row_mut, container_p);
        }
    }
	
    LOG(FATAL, "InsertThread died");
    return 0;
}

void CQueryCmdHandler::WriteXbase(__DataVec &vec,
								  xbase::multi_row_mutator_ptr multi_row_mut,
								  container_ptr container_p)
{
    LOG(INFO, "CQueryCmdHandler::WriteXbase");

	typedef CInsertData::__PeerVec __PeerVec;

	char row[CID_LEN + PEERID_LEN] = {0},value[GCID_LEN + sizeof(U64)];

    size_t add_count = 0;
    for(__DataVec::const_iterator i = vec.begin();i != vec.end();++i)
	{
        CInsertData * p = *i;
        if(!p)
		{
            LOG(ERROR, "CInsertData="<<p<<" from __DataVec");
        }
		else if(p->cid_.length() != CID_LEN)
		{
            LOG(ERROR, "error cid in CInsertData="<<p->ToString()<<", ignore it");
        }
		else if(p->gcid_.length() > GCID_LEN)
		{
            LOG(ERROR, "error gcid in CInsertData="<<p->ToString()<<", ignore it");
        }
		else
		{
            memcpy(row,p->cid_.c_str(),CID_LEN);
			
            if(p->gcid_.empty())
                memset(value,'0',GCID_LEN);
            else
                memcpy(value,p->gcid_.c_str(),p->gcid_.length());

			*reinterpret_cast<U64 *>(value + GCID_LEN) = p->fileSz_;

			__PeerVec::const_iterator j = p->peerId_.begin();
			for(size_t index = 0;j != p->peerId_.end();++j,++index)
			{
                if(j->length() != PEERID_LEN)
				{
                    LOG(ERROR, "error peerid["<<index<<"] in CInsertData="<<p->ToString()<<", ignore it");
                    continue;
                }
				
                memcpy(row + CID_LEN,j->c_str(),PEERID_LEN);

				std::string row_str(row, sizeof row);
				std::string value_str(value, sizeof value);
				LOG(DEBUG, "add row, row_name: {" << dump_string_hex(row_str) << " value: " << dump_string_hex(value_str));

				xbase::row_mutation row_mut(tableInfoPtr_, row_str);
				row_mut.add(familyName_for_ops, value_str);
				container_p->push_back(row_mut.data_ptr());
				
                ++add_count;
            }
        }
		
        CInsertData::PutObject(p);
    }
	
	// 确保整个容器有序
    std::sort(container_p->begin(), container_p->end(), xbase::row_mutation_data_ptr_less);

    LOG(INFO, "add_count = "<<add_count<<", clear __DataVec");
    vec.clear();

    add_count = 0;
	do
	{
        LOG(INFO, "container_p->size() = "<<container_p->size());
        U64 exeTime = 0;
		
        if(dbpStats_)
            exeTime = Tools::GetTimeUs();

		xbase::basic_sync_handler<std::size_t> sh;
		multi_row_mut->async_apply(tableName_, container_p, sh.make_handler());
		sh.wait();
		
        if(dbpStats_)
            exeTime = Tools::GetTimeUs() - exeTime;

		boost::system::error_code ec = sh.get_error_code();
		unsigned succ_row_count = sh.get_succ_row_count();
		
        if(ec)
		{
            LOG(ERROR, "batch write error : "<< ec << " - " << ec.message());
			
			if (succ_row_count>0
				&& succ_row_count<container_p->size())
			{
				container_p = container_ptr(
					new container_type(container_p->begin()+succ_row_count, container_p->end())
					);
			}
			
            if (dbpStats_)
			{
				dbpStats_->insertSuccStats_ += succ_row_count;
				dbpStats_->insertSuccTime_ += exeTime;

                dbpStats_->insertFailStats_ += (container_p->size() - succ_row_count);
                dbpStats_->insertFailTime_ += exeTime;
            }
        }
		else
		{
            LOG(INFO, "write succeeded, " << succ_row_count << " written");
			container_p->clear();

			if (dbpStats_)
			{
				dbpStats_->insertSuccStats_ += succ_row_count;
    	        dbpStats_->insertSuccTime_ += exeTime;
			}
        }
    }while(++add_count < writeRetryTimes_ && container_p->size() >= batWriteCount_);
}

void CQueryCmdHandler::InsureTable()
{
	LOG(DEBUG, "CQueryCmdHandler::InsureTable");

	xbase::table_creation creation(tableName_);

	int compressAlgo = 0;
	if (compressMethodName_ == "QUICKLZ")
	{
		compressAlgo = xbase::compressor::QUICKLZ;
	}
	else if (compressMethodName_ == "ZLIB")
	{
		compressAlgo = xbase::compressor::ZLIB;
	}
	else if (compressMethodName_ == "MINILZO")
	{
		compressAlgo = xbase::compressor::ZLIB;
	}
	else if (compressMethodName_ == "BMZ")
	{
		compressAlgo = xbase::compressor::BMZ;
	}
	else
	{
		compressAlgo = xbase::compressor::NONE;
	}

	xbase::table_creation::group_options g_options;
	g_options.m_block_size = blockSz_;
	g_options.m_cache_rate = cacheRate_;
	g_options.m_max_num_sstable = sstableNum_;
	g_options.m_max_size_sstable = sstableSz_;
	g_options.m_max_size_memtable = memtableSz_;
	g_options.m_compress_method = compressAlgo;

	xbase::table_creation::family_options f_options;
	f_options.m_max_num_version = maxVersion_;
	f_options.m_max_ttl = maxTtl_;
	f_options.m_dead_time = deadTime_;

	creation.add_group(groupName_, g_options);
	creation.add_family(familyName_, groupName_, f_options);

	while (true)
	{
		xbase::table_mutator::sync_handler_type sync_handler;
		clientEnvironmentPtr_->async_apply(creation, sync_handler.make_handler());
		sync_handler.wait();

		boost::system::error_code ec = sync_handler.get_error_code();
		if (ec)
		{
			if (ec == xbase::error::table_already_exists)
			{
				LOG(INFO, "table "<<tableName_<<" already exists");
			}
			else
			{
				LOG(FATAL, "\"create table "<<tableName_<<" ("<<familyName_<<" "
					<<maxVersion_<<" "<<maxTtl_<<" "<<deadTime_<<",group "
					<<groupName_<<" "<<compressMethodName_<<" "<<cacheRate_<<" "<<blockSz_
					<<" "<<sstableNum_<<" "<<sstableSz_<<" "<<memtableSz_
					<<"\" error, errorstr="<<ec.message());

				sleep(3);

				continue;
			}
		}
		else
		{
			LOG(INFO, "create new table "<<tableName_<<" success, sleep "
				<<createTableSleep_<<" s");

			if(createTableSleep_)
				sleep(createTableSleep_);
		}

		// 获得table_info
		xbase::table_info_cache::sync_handler_type sh;	

		LOG(DEBUG,"query table info for " << tableName_);
		clientEnvironmentPtr_->get_table_info_cache().async_get_table_info(tableName_, sh.make_handler());
		sh.wait();

		if(sh.get_error_code())
		{
			LOG(FATAL, "error: " << sh.get_error_code() << " - " << sh.get_error_code().message());
			continue;
		}

		tableInfoPtr_ = sh.get_table_info();
		LOG(INFO, "got table info: " << tableInfoPtr_->to_string());
		break;
	}
}
